package org.wenzi.com.job;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.io.PojoCsvInputFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.wenzi.com.function.WindowResultFunction;
import org.wenzi.com.pojo.UserBehaviorPojo;

import java.io.File;
import java.net.URL;
import java.sql.Timestamp;

/**
 * @author zhaozuowen
 * @date 2021-10-25 19:54
 */
public class PvJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // UserBehavior.csv 的本地文件路径
        URL fileUrl = PvJob.class.getClassLoader().getResource("UserBehavior2.csv");
        Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));
        // 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo
        PojoTypeInfo<UserBehaviorPojo> pojoType = (PojoTypeInfo<UserBehaviorPojo>) TypeExtractor.createTypeInfo(UserBehaviorPojo.class);
        // 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序
        String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
        // 创建 PojoCsvInputFormat
        PojoCsvInputFormat<UserBehaviorPojo> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);

        DataStreamSource<UserBehaviorPojo> input = env.createInput(csvInput,pojoType);
        SingleOutputStreamOperator<UserBehaviorPojo> waterMarkStream = input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehaviorPojo>() {
            @Override
            public long extractAscendingTimestamp(UserBehaviorPojo userBehaviorPojo) {
                return userBehaviorPojo.getTimestamp() * 1000;
            }
        });

        SingleOutputStreamOperator<UserBehaviorPojo> pvStream = waterMarkStream.filter(new FilterFunction<UserBehaviorPojo>() {
            @Override
            public boolean filter(UserBehaviorPojo userBehaviorPojo) throws Exception {
                return userBehaviorPojo.getBehavior().equalsIgnoreCase("pv");
            }
        });

        SingleOutputStreamOperator<Tuple2<Long, Long>> mapStream = pvStream.map(new RichMapFunction<UserBehaviorPojo, Tuple2<Long, Long>>() {
            @Override
            public Tuple2<Long, Long> map(UserBehaviorPojo userBehaviorPojo) throws Exception {
                return Tuple2.of(userBehaviorPojo.getItemId(),1L);
            }
        });

        SingleOutputStreamOperator<Tuple4<Long, Long, Timestamp, Timestamp>> sumStream = mapStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.days(1),Time.hours(1))).
                 trigger(PurgingTrigger.of(ContinuousEventTimeTrigger.of(Time.hours(1)))).
                   reduce((ReduceFunction<Tuple2<Long, Long>>) (t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1),

                        new WindowFunction<Tuple2<Long, Long>, Tuple4<Long, Long, Timestamp, Timestamp>, Tuple, TimeWindow>() {
                            @Override
                            public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<Long, Long>> input, Collector<Tuple4<Long, Long, Timestamp, Timestamp>> out) throws Exception {
                                Tuple2<Long, Long> tuple2 = input.iterator().next();
                                Tuple4<Long, Long, Timestamp, Timestamp> tuple4 = Tuple4.of(tuple2.f0, tuple2.f1, new Timestamp(window.getStart()), new Timestamp(window.getEnd()));
                                out.collect(tuple4);
                            }
                        });
        sumStream.print();

        env.execute("PvJob  execute");
    }
}
